package basicutil.pipeline.core;

import basicutil.pipeline.DataMessage;
import basicutil.pipeline.DataType;

import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class InputPort {
    private final String portId;
    private final DataType dataType;
    private final BlockingQueue<DataMessage> inputQueue;


    AtomicBoolean beConnected = new AtomicBoolean(false);
    OutputPort connectedPort;
    private int maxQueueSize;        // 最大队列大小
enum BlockMode{
    drop,wait,update
}
    public InputPort(String portId, DataType dataType) {
        this.portId = portId;
        this.dataType = dataType;
        this.inputQueue = new LinkedBlockingQueue<>();
        this.maxQueueSize = 1000;
    }


    public boolean canConnect(OutputPort other) {
        return other != null &&
                Objects.equals(this.dataType, other.getDataType());
    }


    public void connectBy(OutputPort other) {
        connectedPort = other;
        beConnected.set(true);
    }

    public DataMessage waitMessage() throws InterruptedException {
        return inputQueue.take();
    }

    public boolean isConnected() {
        return beConnected.get();
    }

    public String getId() {
        return portId;
    }


    public DataType getDataType() {
        return dataType;
    }


    public void setMaxQueueSize(int maxQueueSize) {
        this.maxQueueSize = maxQueueSize;
    }
}

